package cn.getech.data.development.sink.jdbc.config

import cn.getech.data.development.bean.jdbc.{JDBCCollectionConfigBean, JDBCFieldBean}
import cn.getech.data.development.enums.CustomJDBCType.CustomJDBCType
import cn.getech.data.development.enums.{CustomJDBCType, DelimitFormat, OutputFormat}
import cn.getech.data.development.utils.{ConfigurationManager, HiveUtils, JDBCSQLTypeUtils}
import org.apache.flink.table.types.DataType
import org.apache.flink.util.Preconditions


class CustomTableConfig extends Serializable {

  private val config = JDBCConfig()
  /**
   * schema params
   */
  private var params: Array[JDBCFieldBean] = Array[JDBCFieldBean]()
  /**
   * 是否开启动态分区
   */
  var isDynamicPartition: Boolean = false
  /**
   * 是否是分区表
   */
  var isPartitionTable: Boolean = false
  /**
   * 是否开启采集配置
   */
  var isCollectionConfig: Boolean = false
  /**
   * 采集配置
   */
  var collectionConfig: JDBCCollectionConfigBean = null

  /**
   * batch 时间
   */
  var batchInterval: Int = 10
  /**
   * sink方式
   */
  var jdbcType: CustomJDBCType = null
  /**
   * hive
   * 文本格式
   */
  var format: OutputFormat.Value = null
  /**
   * hive
   * 分隔符
   */
  var delimitFormat: DelimitFormat.Value = null


  // jdbc type
  def jdbcType(t: String): CustomTableConfig = {
    Preconditions.checkNotNull(t)
    jdbcType = CustomJDBCType.withName(t)
    jdbcType match {
      case CustomJDBCType.hive => driver("org.apache.hive.jdbc.HiveDriver")
      case CustomJDBCType.impala => driver("org.apache.hive.jdbc.HiveDriver")
      case CustomJDBCType.clickHouse => driver("ru.yandex.clickhouse.ClickHouseDriver")
      case _ => //kudu
    }
    this
  }

  // sql 的参数字段
  def sqlParams(par: Array[JDBCFieldBean]): CustomTableConfig = {
    Preconditions.checkNotNull(par)
    params = par
    this
  }

  // batch的size
  def batchInterval(batchInterval: Int): CustomTableConfig = {
    this.batchInterval = batchInterval
    this
  }

  // hive分区名字段
  def collectionConfig(collectionConfig: JDBCCollectionConfigBean): CustomTableConfig = {
    this.collectionConfig = collectionConfig
    this.isCollectionConfig = collectionConfig != null
    this
  }

  // 动态分区
  def dynamicPartition(isDynamicPartition: Boolean): CustomTableConfig = {
    this.isDynamicPartition = isDynamicPartition
    this
  }

  // 是否是分区表
  def partitionTable(isPartitionTable: Boolean): CustomTableConfig = {
    this.isPartitionTable = isPartitionTable
    this
  }

  // 版本号
  def version(version: String): CustomTableConfig = {
    this.config.version(version)
    this
  }

  // 驱动
  private def driver(driver: String): CustomTableConfig = {
    this.config.driver(driver)
    this
  }

  // url
  def url(url: String): CustomTableConfig = {
    this.config.url(url)
    this
  }

  // 表名
  def tableName(tableName: String): CustomTableConfig = {
    this.config.tableName(tableName)
    this
  }

  // 库名
  def db(db: String): CustomTableConfig = {
    this.config.db(db)
    this
  }

  // 用户名
  def username(username: String): CustomTableConfig = {
    this.config.username(username)
    this
  }

  // 密码
  def password(password: String): CustomTableConfig = {
    this.config.password(password)
    this
  }

  // 格式
  def hdfs_format(format: String): CustomTableConfig = {
    this.format = OutputFormat.withName(format)
    this
  }

  // 分隔符
  def hdfs_delimitFormat(delimitFormat: Int): CustomTableConfig = {
    this.delimitFormat = DelimitFormat.withFormat(delimitFormat)
    this
  }

  // sql解析
    def sqlParser: String = {
      jdbcType match {
        case CustomJDBCType.hive => hiveParser
        case CustomJDBCType.impala => impalaParser
        case CustomJDBCType.clickHouse => clickHouseParser
      }
    }

    private def clickHouseParser: String = {
      val sb = new StringBuilder
      sb.append(s"insert into ${config.db._2}.${config.tablename._2}")
      sb.append(" values (")
      val subSb = new StringBuilder
      for (index <- 0 until params.size) {
        subSb.append("?")
        if (params.size - 1 != index) subSb.append(",")
      }
      sb.append(subSb.toString())
      sb.append(")")
      sb.toString()
    }

    private def impalaParser: String = {
      hiveParser
    }


    private def hiveParser: String = {
      val sb = new StringBuilder
      sb.append(s"insert into ${config.tablename._2}")
      if (isPartitionTable) {
//        sb.append(s" partition(${partitionName}='")
//        sb.append(partitionValue + "') ")
      }
      sb.append(" values (")
      val subSb = new StringBuilder
      for (index <- 0 until params.size) {
        subSb.append("?")
        if (params.size - 1 != index) subSb.append(",")
      }
      sb.append(subSb.toString())
      sb.append(")")
      sb.toString()
    }

  // 获取参数
  def getConfig(): JDBCConfig = config

  def getTableConfig = this


  def getHDFSUri = {
    s"${ConfigurationManager.getProperty("hdfs.hive.warehouse.path")}${config.db._2}.db/${config.tablename._2}/"
  }


  // 字段总数
  def getParamSize = params.size

  def getParams =  params

  def partitionParam(boolean: Boolean) = params.filter(_.isPartition == boolean)

  // 获取对象
  def param_field_names: Array[String] = params.map(_.field_name)

  def param_class_type2Ints: Array[Int] = {
    val utils = new JDBCSQLTypeUtils
    params.map(x => utils.type2Int(x.class_type))
  }


  def param_class_type2DataType: Array[DataType] = {
    val utils = new JDBCSQLTypeUtils
    params.map(x => utils.type2DataType(x.class_type))
  }

  // 获取batch size
  def getBatchInterval = batchInterval


  case class JDBCConfig() {
    private val CONNECTOR_DRIVER = "connector.driver"
    private val CONNECTOR_URL = "connector.url"
    private val CONNECTOR_DB = "connector.databases"
    private val CONNECTOR_TABLENAME = "connector.tablename"
    private val CONNECTOR_USERNAME = "connector.username"
    private val CONNECTOR_PASSWORD = "connector.password"
    private val VERSION = "version"

    var driver: (String, String) = null
    var url: (String, String) = null
    var db: (String, String) = null
    var tablename: (String, String) = null
    var username: (String, String) = null
    var password: (String, String) = null
    var version: (String, String) = null

    def version(version: String): JDBCConfig = {
      Preconditions.checkNotNull(version)
      this.version = (VERSION, version)
      this
    }

    def driver(driver: String): JDBCConfig = {
      Preconditions.checkNotNull(driver)
      this.driver = (CONNECTOR_DRIVER, driver)
      this
    }

    def url(url: String): JDBCConfig = {
      Preconditions.checkNotNull(url)
      this.url = (CONNECTOR_URL, url)
      this
    }

    def tableName(tableName: String): JDBCConfig = {
      Preconditions.checkNotNull(tableName)
      this.tablename = (CONNECTOR_TABLENAME, tableName)
      this
    }

    def db(db: String): JDBCConfig = {
      this.db = (CONNECTOR_DB, db)
      this
    }

    def username(username: String): JDBCConfig = {
      this.username = (CONNECTOR_USERNAME, username)
      this
    }

    def password(password: String): JDBCConfig = {
      this.password = (CONNECTOR_PASSWORD, password)
      this
    }
  }

}
